package com.abyss;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.util.HashMap;
import java.util.Map;

/**
 * 分布式缓存的演示
 */
public class DistributedCacheFileDemo {
    public static void main(String[] args) throws Exception {
        // 1. Env
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 2. Source
        // 2.1 学生的分数source
        DataSource<Tuple3<Integer, String, Integer>> scoreSource = env.fromElements(
                Tuple3.of(1, "数据结构", 99),
                Tuple3.of(2, "英语", 100),
                Tuple3.of(3, "C++", 96),
                Tuple3.of(5, "Java", 97),
                Tuple3.of(3, "Scala", 100)
        );

        // 3. 注册分布式缓存
        env.registerCachedFile("/Users/abyss/Dev/toys/flink/H-flink-learn/src/main/resources/distributed_student.txt", "student");

        // 4. 在map中使用分布式缓存
        MapOperator<Tuple3<Integer, String, Integer>, Tuple3<String, String, Integer>> result = scoreSource.map(new RichMapFunction<Tuple3<Integer, String, Integer>, Tuple3<String, String, Integer>>() {
            private Map<Integer, String> stuMap = new HashMap<>();

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                File cacheFile = getRuntimeContext().getDistributedCache().getFile("student");
                BufferedReader bufferedReader = new BufferedReader(new FileReader(cacheFile));
                String line = null;
                while ((line = bufferedReader.readLine()) != null) {
                    String[] values = line.split(",");
                    stuMap.put(Integer.parseInt(values[0]), values[1]);
                }
            }

            @Override
            public Tuple3<String, String, Integer> map(Tuple3<Integer, String, Integer> value) throws Exception {
                String stuName = this.stuMap.getOrDefault(value.f0, "未知的学生姓名");
                return Tuple3.of(stuName, value.f1, value.f2);
            }
        });

        result.print();
    }
}